package com.liuwei.msmall.search.service.impl;

import com.alibaba.fastjson.JSON;
import com.liuwei.msmall.common.constant.EsConstant;
import com.liuwei.msmall.common.dto.es.EsSkuModelDTO;
import com.liuwei.msmall.search.config.ElasticSearchConfig;
import com.liuwei.msmall.search.service.ProductEsService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.support.IndicesOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.TermQueryBuilder;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Classname ProductEsServiceImpl
 * @Date 2021/4/10 2:59 下午
 * @Created by llww
 */
@Service
@Slf4j
public class ProductEsServiceImpl implements ProductEsService {

    @Resource
    RestHighLevelClient restHighLevelClient;


    @Override
    public Boolean saveToEs(List<EsSkuModelDTO> esSkuModels) throws IOException {
        //批量保存到es中bulk方法 (BulkRequest bulkRequest, RequestOptions options)
        BulkRequest bulkRequest = new BulkRequest();

        for (EsSkuModelDTO esSkuModel : esSkuModels) {
            IndexRequest indexRequest = new IndexRequest(EsConstant.INDEX_NAME);
            indexRequest.id(esSkuModel.getSkuId().toString());
            indexRequest.source(JSON.toJSONString(esSkuModel), XContentType.JSON);
                        bulkRequest.add(indexRequest);
        }

        BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, ElasticSearchConfig.COMMON_OPTIONS);
        //判断批量保存中是否出现异常
        boolean hasFailures = bulk.hasFailures();
        List<String> collect = Arrays.stream(bulk.getItems()).map(item -> {
            if(item.isFailed()){
                log.error("保存到es失败 {}",item.getFailureMessage());
            }
            return item.getId();
        }).collect(Collectors.toList());
        return hasFailures;
    }

    @Override
    public void deleteFromEs(Long spuId) throws IOException {
        DeleteByQueryRequest request = new DeleteByQueryRequest(EsConstant.INDEX_NAME);
        request.setQuery(new TermQueryBuilder("spuId", spuId));
        request.setRefresh(true);
        request.setIndicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
        BulkByScrollResponse response = restHighLevelClient.deleteByQuery(request, ElasticSearchConfig.COMMON_OPTIONS);
        log.info("被删除的文档数："+response.getDeleted());
    }
}
